package com.atguigu.datastream.day05;

import com.atguigu.datastream.bean.Event;
import com.atguigu.datastream.test.day03.Flink_05_Source_useDefault;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;

/**
 * ClassName: Flink05_ProcessTimeWindow_ProcessWindowFun
 * Package: com.atguigu.day05
 * Description:
 *             使用全窗口函数ProcessWindowFun实现UV的计算
 * @Author ChenJun
 * @Create 2023/4/11 18:01
 * @Version 1.0
 */
public class Flink05_ProcessTimeWindow_ProcessWindowFun {
    public static void main(String[] args) throws Exception {
        
        //1. 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.setParallelism(1);

        //2.通过自定义数据源获取数据
        DataStreamSource<Event> streamSource = env.addSource(new Flink_05_Source_useDefault.ClickSource());
        
        //3. 将相同的key放到同一个分区
        KeyedStream<Event, String> keyedStream = streamSource.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event event) throws Exception {
                return "key";
            }
        });

        //4.开启一个基于处理时间的滚动窗口，每五秒计算一次5秒内的UV
        WindowedStream<Event, String, TimeWindow> window = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        //TODO 5.使用全窗口函数ProcessWindowFun实现UV的计算
        
        
        window.process(new ProcessWindowFunction<Event, Integer, String, TimeWindow>() {
            /**
             *
             * @param s   key
             * @param context  上下文对象
             * @param iterable   迭代器 里面放的是窗口中的数据
             * @param collector  收集器，可以将数据发送至下游
             * @throws Exception
             */
            @Override
            public void process(String s, ProcessWindowFunction<Event, Integer, String, TimeWindow>.Context context, Iterable<Event> iterable, Collector<Integer> collector) throws Exception {
                //1.创建HashSet里面存放user，利用hashSet的幂等性去重，以此可以得到UV
                HashSet<String> set = new HashSet<>();
                //2.遍历迭代器 （注意：对于迭代器来说，只能使用一次，因此迭代器底层是链表，当遍历完毕之后，头指针会指向尾结点，因此没办法重复使用，如果想重复使用迭代器中的数据，可以将迭代器中的数据遍历出来放到集合中）
                for (Event event : iterable) {
                    set.add(event.user);
                }
                //3。输出uv
                collector.collect(set.size());
            }
        }).print();

        env.execute();
        
    }
}
